package com.flute.haflute.jobbox.communication;

import java.rmi.RemoteException;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.jobbox.common.CloudContext;
import com.flute.haflute.jobbox.common.ScheduleBase;
import com.flute.haflute.jobbox.common.exception.MessagerException;
import com.flute.haflute.jobbox.communication.messages.MessageBase;

/**
 * 通信平台的后台线程(消息通知器) 接受消息并发送给所有子节点 *
 * 
 */
public class NotifiyDaemon extends ScheduleBase {
	private static Logger logger = LoggerFactory.getLogger(NotifiyDaemon.class);
	private static NotifiyDaemon instance = new NotifiyDaemon();

	/** 所有监听者 */
	private Hashtable<String, Listener> listeners;
	/** 等待发送的消息列表 */
	private Hashtable<String, List<MessageBase>> requests;
	private Object requestsLock = new Object();

	private NotifiyDaemon() {
		super("Notifiy Service", true);
		logger.info("Notifiy Service started");
	}

	public static NotifiyDaemon getInstance() {
		return instance;
	}

	@Override
	public void close() {
	}

	/**
	 * 向主节点注册监听器
	 * 
	 * @see com.flute.haflute.jobbox.communication.NotifyService#register(java.lang.String,
	 *      com.flute.haflute.jobbox.communication.Listener)
	 */
	public void register(String childName, Listener child)
			throws RemoteException {
		if (childName == null)
			throw new RemoteException("host name should not be null");

		// 注册时，向集群添加该节点
		CloudContext.addChild(childName);

		Listener listener = listeners.get(childName);
		try {
			// 检查是否注册的监听器是否已经存在，即子节点是否已经存在于集群中
			if (listener != null && listener.isAlive()) {
				throw new MessagerException("alive");
			}
		} catch (RemoteException e) {
			logger.error("listener for " + childName
					+ " not null, but can't access: ", e);
		} catch (MessagerException e) {
			String errorStr = "listener for " + childName + " already exists";
			logger.error(errorStr);
		}
		listeners.put(childName, child);

		logger.info("regiestered Subscriber one for " + childName);
	}

	/**
	 * 发送消息接口，消息将首先进入等待消息列表中
	 * 
	 * @see com.flute.haflute.jobbox.communication.NotifyService#send(java.lang.String,
	 *      com.flute.haflute.jobbox.communication.messages.MessageBase)
	 */
	public void send(String hostName, MessageBase request)
			throws RemoteException {
		if (hostName == null)
			throw new RemoteException("host name should not be null");

		synchronized (requestsLock) {
			List<MessageBase> oneMessageList = requests.get(hostName);
			if (oneMessageList == null) {
				oneMessageList = new Vector<MessageBase>();
				requests.put(hostName, oneMessageList);
			}
			oneMessageList.add(request);
		}
	}

	/** 向子节点的监听者发送消息 */
	private void doNotify(Map<String, List<MessageBase>> requets4send) {
		logger.info("doNotiy size=" + requets4send.size());
		// Iterator<String> ite = requets4send.keySet().iterator();
		for (Entry<String, List<MessageBase>> entry : requets4send.entrySet()) {
			String childName = entry.getKey();
			List<MessageBase> messages4host = entry.getValue();

			Listener child = listeners.get(childName);
			try {
				child.receive(messages4host);
			} catch (Exception e) {
				logger.error("doNotify error for {}", childName);
				logger.error("doNotify error is", e);
			}
		}// end while
	}

	/**
	 * 调度过程，定期执行
	 * 
	 * @see com.flute.haflute.jobbox.common.ScheduleBase#execute()
	 */
	public void execute() {
		if (requests.size() == 0)
			return;
		Map<String, List<MessageBase>> messages4send = null;
		synchronized (requestsLock) {
			messages4send = requests;
			requests = new Hashtable<String, List<MessageBase>>();
		}
		doNotify(messages4send);
	}

	public void unRegister(String name) throws RemoteException {
		if (name == null)
			return;
		listeners.remove(name);
		CloudContext.delChild(name);

		logger.info("unregistered one listener: {}", name);
	}

	@Override
	protected void setupEnvironment() {
		listeners = new Hashtable<String, Listener>();
		requests = new Hashtable<String, List<MessageBase>>();
	}

	public void broadcast(MessageBase request) throws RemoteException {
		for (String child : listeners.keySet()) {
			try {
				Listener childLister = listeners.get(child);
				childLister.receive(request);
			} catch (RemoteException e) {
				logger.error("doNotify error for {}, child is {}", request,
						child);
				logger.error("doNotify error is", e);
			}
		}

	}

}
